package com.iqiyi.pps.epg.web.timer;

import com.iqiyi.pps.epg.core.model.rtmp.RtmpServer;
import com.iqiyi.pps.epg.core.model.rtmp.RtmpServerPfv;
import com.iqiyi.pps.epg.core.service.rtmp.RtmpService;
import com.iqiyi.pps.epg.core.utils.ApplicationContextUtil;
import org.hibernate.SessionFactory;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RtmpServerPfvSyncTrigger {
    private static Logger logger = LoggerFactory.getLogger(RtmpServerPfvSyncTrigger.class);

    @Autowired
    private RtmpService rtmpService;

    private SessionFactory sessionFactory;
    private Map<String, RtmpServerPfv> rtmpServerPfvMap;
    private Map<Integer, Map<String, RtmpServerPfv>> serverPfvMap;

    public void task() {
        logger.info("[RtmpServerPfvSyncTrigger][task][start]");
        try {
            // 获取所有流服务器列表
            List<RtmpServer> serverList = rtmpService.getServerList();
            sessionFactory = (SessionFactory) ApplicationContextUtil.getBean(SessionFactory.class);

            // 获取所有服务器上的流
            List<RtmpServerPfv> rtmpServerPfvList = rtmpService.getServerPfvList();
            if (rtmpServerPfvList.size() > 0) {
                rtmpServerPfvMap = new HashMap(rtmpServerPfvList.size());
                serverPfvMap = new HashMap(serverList.size());
                for (RtmpServerPfv item : rtmpServerPfvList) {
                    String uniqueToken = getUniqueToken(item);
                    rtmpServerPfvMap.put(uniqueToken, item);
                    if (!serverPfvMap.containsKey(item.getServerId())) {
                        serverPfvMap.put(item.getServerId(), new HashMap(30));
                    }
                    serverPfvMap.get(item.getServerId()).put(uniqueToken, item);
                }
            } else {
                rtmpServerPfvMap = new HashMap();
                serverPfvMap = new HashMap();
            }

            ExecutorService exec = Executors.newFixedThreadPool(serverList.size());
            for (RtmpServer server : serverList) {
                exec.execute(new TaskExecutor(server));
            }
            exec.shutdown();
        } catch (Exception e) {
            logger.info("[RtmpServerPfvSyncTrigger][task][exception=]", e);
        }
    }

    private String getUniqueToken(RtmpServerPfv entity) {
        return entity.getServerId() + "_" + entity.getName() + "_" + entity.getObject();
    }

    private class TaskExecutor implements Runnable {
        private RtmpServer server;

        public TaskExecutor(RtmpServer server) {
            this.server = server;
        }

        @Override
        public void run() {
            try {
                String url = "http://" + server.getPrivateIp() + ":8081/stat";
                logger.info("[RtmpServerPfvSyncTrigger][thread][url={}]", url);
                Document doc = Jsoup.connect(url).get();
                Elements trs = doc.getElementsByTag("tr");
                trs.remove(0);
                for (Element tr : trs) {
                    int i = 0;
                    RtmpServerPfv entity = new RtmpServerPfv();
                    entity.setServerId(server.getId());
                    for (Element td : tr.getElementsByTag("td")) {
                        td.html();
                        String text = td.text();
                        switch (i) {
                            case 0:
                                text = text.substring(text.lastIndexOf("/") + 1);
                                entity.setName(text);
                                break;
                            case 1:
                                entity.setBitrate(Integer.valueOf(text));
                                break;
                            case 2:
                                entity.setDownload(text);
                                break;
                            case 3:
                                entity.setDownloadTotal(text);
                                break;
                            case 4:
                                entity.setDownloadErrorCode(text);
                                break;
                            case 5:
                                entity.setDownloadRetryTimes(Integer.valueOf(text));
                                break;
                            case 6:
                                entity.setDownloadDuration(text);
                                break;
                            case 7:
                                entity.setObject(text);
                                break;
                            case 8:
                                entity.setUpload(text);
                                break;
                            case 9:
                                entity.setUploadTotal(text);
                                break;
                            case 10:
                                entity.setUploadErrorCode(text);
                                break;
                            case 11:
                                entity.setUploadRetryTimes(Integer.valueOf(text));
                                break;
                            case 12:
                                entity.setUploadDuration(text);
                                break;
                            case 13:
                                entity.setMirror(text);
                                break;
                        }
                        ++i;
                    }
                    // 判断是否新增
                    String currentUniqueToken = getUniqueToken(entity);
                    if (rtmpServerPfvMap.containsKey(currentUniqueToken)) {
                        RtmpServerPfv temp = rtmpServerPfvMap.get(currentUniqueToken);
                        entity.setId(temp.getId());
                        sessionFactory.openSession().evict(temp);
                        rtmpServerPfvMap.remove(currentUniqueToken);
                        serverPfvMap.get(temp.getServerId()).remove(currentUniqueToken);
                    }
                    entity.setStatus(RtmpServerPfv.STATUS_VALID);
                    rtmpService.saveServerPfv(entity);
                }

                // 处理失效的
                if (serverPfvMap.containsKey(server.getId()) && serverPfvMap.get(server.getId()).size() > 0) {
                    for (RtmpServerPfv item : serverPfvMap.get(server.getId()).values()) {
                        if (item.getStatus() == RtmpServerPfv.STATUS_VALID) {
                            item.setStatus(RtmpServerPfv.STATUS_INVALID);
                            rtmpService.saveServerPfv(item);
                        }
                    }
                }
            } catch (Exception e) {
                logger.info("[RtmpServerPfvSyncTrigger][thread][exception=]", e);
            }
        }
    }
}
